package com.garden.rocketmq.learn.demo.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/** 过滤生产者 生成发送含过滤属性消息
 * Created by Garden on 2020/8/21.
 */
public class FilterProducer {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException, MQBrokerException {
        //实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("filter_producer_group");
        //设置NameServer的地址，Producer启动后通过NameServer获取目标Topic的路由信息
        producer.setNamesrvAddr("localhost:9876");
        //启动Producer实例
        producer.start();

        //发送含tag消息
        //tag_a
        for (int i = 0; i < 5; i++){
            //创建消息，指定主题Topic，标签Tag，业务标识Key，消息体MsgBody [Tag可用于消息过滤，Key可用于消息查询]
            Message msg = new Message("test_topic","filter_producer_tag_a"
                    ,"filter_producer_key_a." + i, ("FilterProducer Msg a." + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //发送消息至Broker
            SendResult sendResult = producer.send(msg);
            //通过发送结果查看是否发送成功
            System.out.printf("%s%n", sendResult);
        }
        //tag_b
        for (int i = 0; i < 5; i++){
            //创建消息，指定主题Topic，标签Tag，业务标识Key，消息体MsgBody [Tag可用于消息过滤，Key可用于消息查询]
            Message msg = new Message("test_topic","filter_producer_tag_b"
                    ,"filter_producer_key_b." + i, ("FilterProducer Msg b." + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //发送消息至Broker
            SendResult sendResult = producer.send(msg);
            //通过发送结果查看是否发送成功
            System.out.printf("%s%n", sendResult);
        }

        //发送含消息属性消息
        //property = 1
        //创建消息，指定主题Topic，标签Tag，业务标识Key，消息体MsgBody [Tag可用于消息过滤，Key可用于消息查询]
        Message msg1 = new Message("test_topic","filter_producer_tag"
                ,"filter_producer_key." + 1, ("FilterProducer Msg " + 1).getBytes(RemotingHelper.DEFAULT_CHARSET));
        //设置消息属性
        msg1.putUserProperty("property","1");
        //发送消息至Broker
        SendResult sendResult1 = producer.send(msg1);
        //通过发送结果查看是否发送成功
        System.out.printf("%s%n", sendResult1);
        //property = 2
        //创建消息，指定主题Topic，标签Tag，业务标识Key，消息体MsgBody [Tag可用于消息过滤，Key可用于消息查询]
        Message msg2 = new Message("test_topic","filter_producer_tag"
                ,"filter_producer_key." + 2, ("FilterProducer Msg " + 2).getBytes(RemotingHelper.DEFAULT_CHARSET));
        //设置消息属性
        msg2.putUserProperty("property","2");
        //发送消息至Broker
        SendResult sendResult2 = producer.send(msg2);
        //通过发送结果查看是否发送成功
        System.out.printf("%s%n", sendResult2);


    }

}
